package com.dgw.springbootkafka.sender;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.websocket.SendResult;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program: springboot-kafka
 * @description:
 * @author: Mr.Dai
 * @create: 2020-02-22 23:01
 **/
@Component
public class ASynProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    private AtomicInteger i = new AtomicInteger(0);

    //@Scheduled(fixedRate = 2000, initialDelay = 5000)
    public void send() {
        i.incrementAndGet();
        // 三个参数，topic,key：用户分配partition,value:发送的值
        ProducerRecord<String, String> record = new ProducerRecord<>("test-syn", "topic_" + i.get(), "test-syn-" + i.get());

        kafkaTemplate.send(record).addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable throwable) {
                throwable.printStackTrace();
                System.out.println("发送失败!");
            }

            @Override
            public void onSuccess(Object o) {
                System.out.println("发送成功!");
            }
        });
        kafkaTemplate.flush();
    }
}
